package lyw.app.dwd;

import lyw.app.BaseSqlApp;
import lyw.common.Constant;
import lyw.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdTradeUserOrder extends BaseSqlApp {
    public static void main(String[] args) {
        new DwdTradeUserOrder().init(3002, 2, "DwdTradeUserOrder");
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {

        // 1. 读取ods_db数据
        readOdsDb(tEnv, "DwdTradeUserOrder");
//        tEnv.sqlQuery(" select * from ods_db").execute().print();
        // 2. 过滤分类表
        Table baseCategoryInfo = tEnv.sqlQuery(
                    "select " +
                        " data['id'] id, " +
                        " data['category_name'] category_name " +
                        " from ods_db " +
                        " where `database`='edu' " +
                        " and `table`='base_category_info' " +
                        " and `type`='insert'"
        );
        baseCategoryInfo.execute().print();

        tEnv.createTemporaryView("base_category_info", baseCategoryInfo);
        // 3. 过滤学科表
        Table baseSubjectInfo = tEnv.sqlQuery(
                "select " +
                        "data['id'] id, " +
                        "data['subject_name'] subject_name, " +
                        "data['category_id'] category_id " +
                        "from ods_db " +
                        "where `database`='edu' " +
                        "and `table`='base_subject_info' " +
                        "and `type`='insert'"
        );
        tEnv.createTemporaryView("base_subject_info",baseSubjectInfo);
        
        // 4. 过滤课程表
        Table courseInfo = tEnv.sqlQuery(
                "select " +
                        " data['id'] id, " +
                        " data['course_name'] course_name, " +
                        " data['subject_id'] subject_id " +
                        " from ods_db " +
                        " where `database`='edu' " +
                        " and `table`='course_info' " +
                        " and `type`='insert'"
        );
        tEnv.createTemporaryView("course_info", courseInfo);
//courseInfo.execute().print();
        // 5. 过滤订单明细表
        Table orderDetail = tEnv.sqlQuery(
                "select " +
                        " data['course_id'] course_id, " +
                        " data['order_id'] order_id, " +
                        " data['user_id'] user_id, " +
                        " data['final_amount'] final_amount " +
                        " from ods_db " +
                        " where `database`='edu' " +
                        " and `table`='order_detail' " +
                        " and `type`='insert'"
        );
        tEnv.createTemporaryView("order_detail", orderDetail);

        // 6. 类别 学科 课程 订单详细 4张表进行join
        Table result = tEnv.sqlQuery(
                "select " +
                        "od.order_id id, " +
                        "od.user_id, " +
                        "bc.id category_id, " +
                        "bc.category_name, " +
                        "bs.id subject_id, " +
                        "bs.subject_name, " +
                        "ci.id course_id, " +
                        "ci.course_name, " +
                        "od.final_amount " +
                        "from base_category_info bc " +
                        "join base_subject_info bs on bc.id=bs.category_id " +
                        "join course_info ci on bs.id=ci.subject_id " +
                        "join order_detail od on ci.id=od.course_id"
        );

        // 7. 写出到kafka中
        tEnv.executeSql(
                "create table dwd_trade_user_order( " +
                        "id string, " +
                        "user_id string, " +
                        "category_id string, " +
                        "category_name string, " +
                        "subject_id string, " +
                        "subject_name string, " +
                        "course_id string, " +
                        "course_name string, " +
                        "final_amount string " +
                        ")" + SQLUtil.getKafkaSink(Constant.TOPIC_DWD_TRADE_USER_ORDER)
        );
//        result.execute().print();
//        result.executeInsert("dwd_trade_user_order");
    }
}
